package com.jd.cmdb.tsc.zk.service;

import com.jd.cmdb.tsc.service.TaskLogService;
import com.jd.cmdb.tsc.service.TaskService;
import com.jd.cmdb.tsc.zk.exception.ZKExceptionHandler;
import com.jd.cmdb.tsc.zk.factory.ClientFactoryBean;
import com.jd.cmdb.tsc.zk.listener.TSCNodeListener;
import com.jd.cmdb.tsc.zk.listener.WorkerNodeListener;
import com.jd.cmdb.tsc.zk.listener.LeaderSelListener;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.leader.LeaderSelector;
import org.apache.curator.framework.recipes.leader.LeaderSelectorListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;

/**
 * Created by tianzhengguo on 2016/1/15.
 */
@Service("zkService")
public class DefaultZKServiceImpl implements ZKService {

    private static final Logger logger = LoggerFactory.getLogger(DefaultZKServiceImpl.class);

    @Resource
    private ClientFactoryBean factory;

    @Resource
    private CuratorFramework client;

    @Resource
    private TaskService taskService;

    @Resource
    private TaskLogService taskLogService;

    @Value("${zk.tscnode}")
    private String tscNode;
    @Value("${zk.workernode}")
    private String workerNode;

    private LeaderSelector leaderSelector;
    private PathChildrenCache workerNodeWatcher;
    private NodeCache tscNodeWatcher;
    private CountDownLatch tscNodeLatch;

    @Override
    @PostConstruct
    public void init() {
        try {
            client = factory.createInstance();
            tscNodeLatch = new CountDownLatch(1);
            start();
            selLeader();
            addWatcherListener();
            addLeaderWatchListener();
        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
        }
    }

    @Override
    public boolean isLeader() {
        try {
            return leaderSelector.hasLeadership();
        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
            return false;
        }
    }

    @Override
    public void selLeader() {
        try {
            if (!isExist(tscNode)){
                client.create().forPath(tscNode);
            }
            LeaderSelectorListener leaderListener = new LeaderSelListener(tscNodeLatch,this);
            leaderSelector = new LeaderSelector(client, tscNode, leaderListener);
            leaderSelector.autoRequeue();
            leaderSelector.setId(UUID.randomUUID().toString());
            leaderSelector.start();
            if(!leaderSelector.getLeader().isLeader()) {
                logger.info("Leader选举中...");
                while (!leaderSelector.getLeader().isLeader()) ;
                logger.info("Leader选举结束,领导id为:{}",leaderSelector.getLeader().getId());
            }
        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
        }
    }

    public void addLeaderWatchListener() {
        try {
            if(isLeader())
                return ;
            if (!isExist(tscNode)) {
                client.create().forPath(tscNode);
            }
            NodeCacheListener listener = new TSCNodeListener(this,tscNodeLatch,tscNode);

            tscNodeWatcher = new NodeCache(client,tscNode);
            tscNodeWatcher.getListenable().addListener(listener);
            tscNodeWatcher.start();
        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
        }
    }

    @Override
    public void addWatcherListener() {
        try {
            if (!isExist(workerNode))
                client.create().forPath(workerNode);
            WorkerNodeListener watcherListener =
                    new WorkerNodeListener(taskService, taskLogService, this, tscNodeLatch);
            workerNodeWatcher = new PathChildrenCache(client, workerNode, true);
            workerNodeWatcher.getListenable().addListener(watcherListener);
            workerNodeWatcher.start();

        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
        }
    }

    @Override
    public boolean isExist(String path) {
        try {
            return null != client.checkExists().forPath(path);
        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
            return false;
        }
    }

    @Override
    public void delete(String path) {
        try {
            client.delete().forPath(path);
        } catch (Exception e) {
            ZKExceptionHandler.handlerZKException(e);
        }
    }

    @Override
    public void start() {
        try {
            if (!client.isStarted())
                client.start();
        } catch (Exception e) {
            logger.error("客户端启动失败.");
        }
    }

    @Override
    public void close() {
        client.close();
    }

    @Override
    public String getTSCNode() {
        return tscNode;
    }

    @Override
    public String getWorkerNode() {
        return workerNode;
    }

    @Override
    public LeaderSelector getLeaderSelector() {
        return leaderSelector;
    }

    @Override
    public PathChildrenCache getWorkerNodeWatcher() {
        return workerNodeWatcher;
    }

    @Override
    public NodeCache getTSCNodeWatcher() {
        return tscNodeWatcher;
    }

}
